package com.xzx.consume;

import com.xzx.common.bean.CallLog;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.NamespaceExistException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2024/8/10 13:52
 */
public class HbaseDao {
    private static final byte[] family = "d".getBytes();
    private Connection connection;

    public HbaseDao() throws IOException {
        Configuration hBaseConfiguration = HBaseConfiguration.create();
        hBaseConfiguration.set("hbase.zookeeper.quorum", "localhost");
        hBaseConfiguration.set("hbase.zookeeper.port", "2181");
        hBaseConfiguration.set("hbase.client.ipc.pool.size", "120");
        hBaseConfiguration.set("hbase.rpc.timeout", "180000");//3 秒,默认是1秒
        connection = ConnectionFactory.createConnection(hBaseConfiguration);
    }

    public void createTable() {
        try (Admin admin = connection.getAdmin()) {
            createNamespaceIfNotExists(admin, "xzx");
            TableName tableName = TableName.valueOf("xzx:call_log");
            if (admin.tableExists(tableName)) {
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            }
            if (!admin.tableExists(tableName)) {
                admin.createTable(TableDescriptorBuilder.newBuilder(tableName)
                        .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("d"))
                                .setMaxVersions(3)
                                .setCompressionType(Compression.Algorithm.SNAPPY)
                                .build())
                        .build(), getSplitKeys(3));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public byte[][] getSplitKeys(int regionNumbers) {
        byte[][] splitKeys = new byte[regionNumbers-1][];
        for (int i = 1; i < regionNumbers; i++) {
            splitKeys[i-1] = Bytes.toBytes(i +"|");
        }
        return splitKeys;
    }

    private void createNamespaceIfNotExists(Admin admin, String ns) throws IOException {
        try {
            admin.createNamespace(NamespaceDescriptor.create("xzx").build());
        } catch (NamespaceExistException e) {
            System.out.println("已经存在，不用创建");
        }
    }

    public void put(CallLog callLog) throws IOException {
        try(Table table = connection.getTable(TableName.valueOf("xzx:call_log"))){
            String rowKey = getRowKey(callLog);
            Put put = new Put(rowKey.getBytes());
            put.addColumn(family, "caller".getBytes(), callLog.getCaller().getBytes());
            put.addColumn(family, "callee".getBytes(), callLog.getCallee().getBytes());
            put.addColumn(family, "callTs".getBytes(), callLog.getCallStartTime().getBytes());
            put.addColumn(family, "duration".getBytes(), Bytes.toBytes(callLog.getDuration()));
            put.addColumn(family, "isCaller".getBytes(), Bytes.toBytes("1"));
            table.put(put);
        }
    }

    private String getRowKey(CallLog callLog) {
        int regionNumber = (callLog.getCaller() + callLog.getCallStartTime().substring(0, 6)).hashCode() % 3;
        return regionNumber + "|" + callLog.getCaller() + "_" + callLog.getCallStartTime().substring(0, 6);
    }
}
